﻿#include "EventLoop.h"
#include <assert.h>
#include <sys/socket.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>

struct EventLoop* eventLoopInit()
{
	return eventLoopInitEx(NULL);
}

// 写数据，唤醒阻塞的子线程
void taskWakeup(struct EventLoop* evLoop)
{
	const char* msg = "我是要成为海贼王的男人!!!";//此处随便一个非空字符串即可
	write(evLoop->socketPair[0], msg, strlen(msg));//向文件描述符写数据，解除IO复用阻塞
}

// 读数据
int readLocalMessage(void* arg)
{
	struct EventLoop* evLoop = (struct EventLoop*)arg;
	char buf[256];
	read(evLoop->socketPair[1], buf, sizeof(buf));
	return 0;
}

struct EventLoop* eventLoopInitEx(const char* threadName)
{
	struct EventLoop* evLoop = (struct EventLoop*)malloc(sizeof(struct EventLoop));
	evLoop->isQuit = false;
	evLoop->threadID = pthread_self();//获取当前线程id（主线程一次，子线程四次）
	pthread_mutex_init(&evLoop->mutex, NULL);//初始化互斥锁
	strcpy(evLoop->threadName, threadName == NULL ? "MainThread" : threadName);//判断是不是主线程，主线程调用时给NULL参数
	//evLoop->dispatcher = &SelectDispatcher;//获取具体实例 ，定义在头文件中
	evLoop->dispatcher = &EpollDispatcher;//获取具体实例 ，定义在头文件中
	evLoop->dispatcherData = evLoop->dispatcher->init();//初始化数据
	// 链表 头尾结点初始化
	evLoop->head = evLoop->tail = NULL;
	// map
	evLoop->channelMap = channelMapInit(128);//设置初始大小
	int ret = socketpair(AF_UNIX, SOCK_STREAM, 0, evLoop->socketPair);//创建本地通信文件描述符 传出长度为2的int数组(装两个本地通信的文件描述符)
	if (ret == -1)
	{
		perror("socketpair");
		exit(0);
	}
	/*将用于主线程 解除子线程 select poll epoll（IO复用模型） 阻塞的本地通信的文件描述符 加到任务队列中
	* "安插卧底"，解除阻塞
	*/

	// 指定规则: evLoop->socketPair[0] 发送数据, evLoop->socketPair[1] 接收数据
	struct Channel* channel = channelInit(evLoop->socketPair[1], ReadEvent,
		readLocalMessage, NULL, NULL, evLoop);//回调函数指定为 空 ，参数指定为 空
	// channel 添加到任务队列
	eventLoopAddTask(evLoop, channel, ADD);//最终会添加到Dispatcher（IO复用模型）中

	return evLoop;
}

int eventLoopRun(struct EventLoop* evLoop)
{
	assert(evLoop != NULL);
	// 取出事件分发和检测模型
	struct Dispatcher* dispatcher = evLoop->dispatcher;
	// 比较线程ID是否正常，每一个eventloop对应一个线程
	if (evLoop->threadID != pthread_self())
	{
		return -1;
	}
	// 循环进行事件处理
	while (!evLoop->isQuit)//反应堆没有退出就一直循环
	{
		dispatcher->dispatch(evLoop, 2);    // 将子线程的eventloop发送给对应的IO检测模型，阻塞检测 超时时长 2s，并调用处理函数，完成数据处理
		eventLoopProcessTask(evLoop);//处理任务队列，任务分发给Dispatcher
	}
	return 0;
}

int eventActivate(struct EventLoop* evLoop, int fd, int event)
{
	if (fd < 0 || evLoop == NULL)
	{
		return -1;
	}
	// 取出channel
	struct Channel* channel = evLoop->channelMap->list[fd];
	assert(channel->fd == fd);
	if (event & ReadEvent && channel->readCallback)//事件相同，回调函数非空
	{
		channel->readCallback(channel->arg);//调用回调函数
	}
	if (event & WriteEvent && channel->writeCallback)
	{
		channel->writeCallback(channel->arg);
	}
	return 0;
}
//添加任务到Eventloop模型的  更新任务 队列

int eventLoopAddTask(struct EventLoop* evLoop, struct Channel* channel, int type)
{
	// 加锁, 保护共享资源
	pthread_mutex_lock(&evLoop->mutex);
	// 创建新节点
	struct ChannelElement* node = (struct ChannelElement*)malloc(sizeof(struct ChannelElement));
	node->channel = channel;
	node->type = type;
	node->next = NULL;

	if (evLoop->head == NULL)//链表为空
	{
		evLoop->head = evLoop->tail = node;//加入队列
	}
	else
	{
		evLoop->tail->next = node;  // add 尾插法
		evLoop->tail = node;        // 后移
	}
	pthread_mutex_unlock(&evLoop->mutex);
	// 处理节点
	/*
	* 细节://假设当前Eventloop反应堆属于子线程
	*   1. 对于链表节点的添加: 可能是当前线程也可能是其他线程(主线程)
	*       1). 修改fd的事件, 当前子线程发起, 当前子线程处理
	*       2). 添加新的fd, 添加任务节点的操作是由主线程发起的
	*   2. 不能让主线程处理任务队列, 需要由当前的子线程取处理
	*/
	if (evLoop->threadID == pthread_self())
	{
		// 当前子线程(基于子线程的角度分析)
		eventLoopProcessTask(evLoop);
	}
	else
	{
		// 主线程 -- 告诉子线程处理任务队列中的任务
		// 1. 子线程在工作 2. 子线程被阻塞了:select, poll, epoll
		taskWakeup(evLoop);//手动唤醒阻塞的 select poll epoll，创建一个专用的文件描述符，并使其就绪，阻塞解除
	}
	return 0;
}

// 管理任务 ，将当前  更新任务队列 更新到IO复用模型检测的channel队列中  最后 更新任务队列 为空
int eventLoopProcessTask(struct EventLoop* evLoop)
{
	//互斥资源上锁
	pthread_mutex_lock(&evLoop->mutex);

	// 取出头结点，头插法
	struct ChannelElement* head = evLoop->head;
	while (head != NULL)
	{
		//获得指向channel的指针
		struct Channel* channel = head->channel;
		if (head->type == ADD)//向任务队列添加任务
		{
			// 添加
			eventLoopAdd(evLoop, channel);
		}
		else if (head->type == DELETE)//删除任务队列里的任务
		{
			// 删除
			eventLoopRemove(evLoop, channel);
		}
		else if (head->type == MODIFY)//修改任务队列里的任务
		{
			// 修改
			eventLoopModify(evLoop, channel);
		}
		struct ChannelElement* tmp = head;
		head = head->next;//队列头指针后移
		free(tmp);//释放已用完数据
	}
	evLoop->head = evLoop->tail = NULL;//
	pthread_mutex_unlock(&evLoop->mutex);
	return 0;
}
//任务添加到IO复用检测集合中
int eventLoopAdd(struct EventLoop* evLoop, struct Channel* channel)
{
	int fd = channel->fd;
	struct ChannelMap* channelMap = evLoop->channelMap;
	if (fd >= channelMap->size)//下标对应文件描述符，大于等于size 则map中没有该文件描述符
	{
		// 没有足够的空间存储键值对 fd - channel ==> 扩容
		if (!makeMapRoom(channelMap, fd, sizeof(struct Channel*)))//fd文件描述符 作为下标 正好装入该文件描述符
		{
			return -1;
		}
	}
	// 找到fd对应的数组元素位置, 并存储
	if (channelMap->list[fd] == NULL)
	{
		channelMap->list[fd] = channel;
		evLoop->dispatcher->add(channel, evLoop);//添加到检测集合中
	}
	return 0;
}
//任务从IO复用检测集合中删除
int eventLoopRemove(struct EventLoop* evLoop, struct Channel* channel)
{
	int fd = channel->fd;
	struct ChannelMap* channelMap = evLoop->channelMap;
	if (fd >= channelMap->size)//不在检查集合中
	{
		return -1;
	}
	int ret = evLoop->dispatcher->remove(channel, evLoop);//从IO监测集合中删除
	return ret;
}
//更改IO检测集合中的任务
int eventLoopModify(struct EventLoop* evLoop, struct Channel* channel)
{
	int fd = channel->fd;
	struct ChannelMap* channelMap = evLoop->channelMap;
	if (channelMap->list[fd] == NULL)//任务不在检测集合中
	{
		return -1;
	}
	int ret = evLoop->dispatcher->modify(channel, evLoop);//修改
	return ret;
}
//销毁channel
int destroyChannel(struct EventLoop* evLoop, struct Channel* channel)
{
	// 删除 channel 和 fd 的对应关系
	evLoop->channelMap->list[channel->fd] = NULL;
	// 关闭 fd
	close(channel->fd);
	// 释放 channel
	free(channel);
	return 0;
}